package com.app.server;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CopyOnWriteArraySet;
import javax.annotation.Resource;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;

import com.app.conf.RedisService;
import com.app.util.CommonHelp;
import com.app.util.SpringUtils;
import com.app.vo.WebSocketVO;
import lombok.extern.slf4j.Slf4j;
import net.sf.json.JSONObject;
import org.springframework.stereotype.Component;

/**
 * @author： N
 * @date： 2022/5/23 16:27
 * @Description： WebSocket操作类
 */
@ServerEndpoint("/websocket/{userId}")
@Component
@Slf4j
public class WebSocketSever {

    // 与某个客户端的连接会话，需要通过它来给客户端发送数据
    private Session session;
    @Resource
    private RedisService redisService = SpringUtils.getBean(RedisService.class);

    // session集合,存放对应的session
    private static ConcurrentHashMap<Integer, Session> sessionPool = new ConcurrentHashMap<>();

    // concurrent包的线程安全Set,用来存放每个客户端对应的WebSocket对象。
    private static CopyOnWriteArraySet<WebSocketSever> webSocketSet = new CopyOnWriteArraySet<>();

    /**
     * 建立WebSocket连接
     *
     * @param session
     * @param userId  用户ID
     */
    @OnOpen
    public void onOpen(Session session, @PathParam(value = "userId") Integer userId) {
        log.info("WebSocket建立连接中,连接用户ID：{}", userId);
        try {
            Session historySession = sessionPool.get(userId);
            // historySession不为空,说明已经有人登陆账号,应该删除登陆的WebSocket对象
            if (historySession != null) {
                webSocketSet.remove(historySession);
                historySession.close();
            }
        } catch (IOException e) {
            log.error("重复登录异常,错误信息：" + e.getMessage(), e);
        }
        // 建立连接
        this.session = session;
        webSocketSet.add(this);
        sessionPool.put(userId, session);
        log.info("建立连接完成,当前在线人数为：{}", webSocketSet.size());
    }

    /**
     * 发生错误
     *
     * @param throwable e
     */
    @OnError
    public void onError(Throwable throwable) {

        throwable.printStackTrace();
    }

    /**
     * 连接关闭
     */
    @OnClose
    public void onClose() {
        webSocketSet.remove(this);

        log.info("连接断开,当前在线人数为：{}", webSocketSet.size());
    }

    /**
     * 接收客户端消息
     * action:1主播2观众
     * code:1连接2心跳
     *
     * @param message 接收的消息
     */
    @OnMessage
    public void onMessage(String message) {
        log.info("收到客户端发来的消息：{}", message);
        JSONObject json_msg = JSONObject.fromObject(message);
        int action = (int) json_msg.get("action");
        int code = (int) json_msg.get("code");
        int userId = (int) json_msg.get("userId");
        String roomId = (String) json_msg.get("roomId");
        WebSocketVO vo = new WebSocketVO();
        vo.setRoomId(roomId);
        vo.setAction(action);
        vo.setUserId(userId);
        vo.setHeartbeatTime(CommonHelp.getNowTimestamp());
        if (code == 1) {

            List<WebSocketVO> webSocketVOList = null;

            webSocketVOList = (List<WebSocketVO>) redisService.getStringKey(roomId);

            //建立连接时发送消息，直播间redis加入观众或主播
            if (webSocketVOList != null && webSocketVOList.size() > 0) {

                //查询重复，删除redis中重复数据
                for (int i = 0; i < webSocketVOList.size(); i++) {

                    if (webSocketVOList.get(i).getUserId() == userId) {
                        webSocketVOList.remove(i);
                    }
                }
                webSocketVOList.add(vo);

                redisService.deleteKey(roomId);
                redisService.setStringKey(roomId, webSocketVOList);
            } else {
                webSocketVOList = new ArrayList<>();
                webSocketVOList.add(vo);
                redisService.setStringKey(roomId, webSocketVOList);
            }

            JSONObject json = new JSONObject();
            json.put("number", webSocketVOList.size() - 1);
            //直播redis数据长度0时为直播关闭
            if (webSocketVOList.size() > 0) {
                json.put("state", 1);
            } else {
                json.put("state", 0);
            }
            //给直播间所有发送人数和状态
//            for (int i = 0; i < webSocketVOList.size(); i++) {
//                if (webSocketVOList.get(i).getRoomId().equals(roomId)){
//                    sendMessageByUser(webSocketVOList.get(i).getUserId(),json.toString());
//                }
//            }
            sendMessageByUser(userId,json.toString());
        } else if (code == 2)//心跳逻辑
        {

            List<WebSocketVO> webSocketVOList = (List<WebSocketVO>) redisService.getStringKey(roomId);
            for (int i = 0; i < webSocketVOList.size(); i++) {
                //查询到正在心跳用户更新心跳时间
                if (roomId.equals(webSocketVOList.get(i).getRoomId()) && userId == webSocketVOList.get(i).getUserId()) {
                    webSocketVOList.remove(i);
                    webSocketVOList.add(vo);
                    redisService.deleteKey(roomId);
                    redisService.setStringKey(roomId, webSocketVOList);

                }
            }
            JSONObject json = new JSONObject();
            json.put("number", webSocketVOList.size() - 1);
            //直播redis数据长度0时为直播关闭
            json.put("state", 1);
            sendMessageByUser(userId, json.toString());
        }

    }

    /**
     * 推送消息到指定用户
     *
     * @param userId  用户ID
     * @param message 发送的消息
     */
    public static void sendMessageByUser(Integer userId, String message) {
        log.info("用户ID：" + userId + ",推送内容：" + message);
        Session session = sessionPool.get(userId);
        try {

            if (session != null && session.isOpen()) {
                session.getBasicRemote().sendText(message);
            }
        } catch (IOException e) {
            log.error("推送消息到指定用户发生错误：" + e.getMessage(), e);
        }
    }

    /**
     * 群发消息
     *
     * @param message 发送的消息
     */
    public static void sendAllMessage(String message) {
        log.info("发送消息：{}", message);
        for (WebSocketSever webSocket : webSocketSet) {
            try {
                webSocket.session.getBasicRemote().sendText(message);
            } catch (IOException e) {
                log.error("群发消息发生错误：" + e.getMessage(), e);
            }
        }
    }

    /**
     * 判断连接是否存在
     *
     * @param userId 发送的消息
     */
    public boolean isOpen(String userId) {
        return sessionPool.get(userId).isOpen();
    }

}

